package com.shujia.spark.sql

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo02SourceAPI {

  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local")
      // 在SparkSQL中如果出现了shuffle操作，默认会生成200个分区，即对应200个任务
      .config("spark.sql.shuffle.partitions", "2") // 默认为200
      .getOrCreate()

    // 导入SparkSession提供的一些隐式转换，可以完成RDD与DF之间的转换
    import spark.implicits._

    // 1、csv：逗号分隔符文件，常用于读取文本文件
    val stuDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id String,name String,age Int,gender String,clazz String")
      .load("spark/data/stu/students.txt")

    // 同样在SparkSQL中也需要”行为算子“触发任务的执行
    // 在SparkSQL中常用的行为算子一般有两种：1、打印 2、保存

    /**
     * 1、打印：不同于RDD需要使用foreach加上println进行打印
     * 在SparkSQL中可以直接使用show方法来对数据进行展示
     * show方法可以接收三个参数：
     * numRows指定打印的数据条数
     * truncate指定是否将数据截断再进行展示
     * vertical执行是否水平还是垂直展示数据，默认false表示水平展示
     */
    stuDF.show() // 默认展示20条，并将truncate设为true
    stuDF.show(10)
    stuDF.show(numRows = 15, truncate = false) // 如果某个数据比较长，想查看完整的数据则需要将truncate置为false
    // vertical表示是否以垂直的方式进行展示，默认为false，一般不会使用
    stuDF.show(numRows = 15, truncate = 1, vertical = true)
    stuDF.show(numRows = 15, truncate = 0, vertical = true)

    // 使用foreach打印
    stuDF
      .foreach(row => {
        // Row对象表示带结构的一行数据，可以通过列名获取对应的列值
        val id: String = row.getAs[String]("id")
        val name: String = row.getAs[String]("name")
        val age: Int = row.getAs[Int]("age")
        val gender: String = row.getAs[String]("gender")
        val clazz: String = row.getAs[String]("clazz")
        println(s"$id,$name,$age,$gender,$clazz")
      })

    /**
     * 2、保存数据：通过DF调用write进行保存
     * Spark SQL在保存的数据的时候提供了多种写入方式，可以通过mode使用SaveMode进行指定
     * SaveMode总共提供了四种不同的写入方式：
     * Append：追加
     * Overwrite：覆盖
     * ErrorIfExists：存在即报错
     * Ignore：存在即不写入
     */
    stuDF.write.format("json").mode(SaveMode.Overwrite)
    //      .save("spark/data/stu/json/")
    stuDF.write.format("orc").mode(SaveMode.Overwrite)
    //      .save("spark/data/stu/orc/")
    stuDF.write.format("parquet").mode(SaveMode.Overwrite)
    //      .save("spark/data/stu/parquet/")


    // 2、json格式：读取的时候不需要指定结构，但存储空间有点浪费
    val jsonStuDF: DataFrame = spark
      .read
      .format("json")
      // 由于json格式每一条数据都自带格式，所以不需要再指定Schema了
      .load("spark/data/stu/json/")

    //    jsonStuDF.show()

    // 3、orc格式：同样在读取时不需要指定结构，由于使用了snappy进行了压缩，所以空间利用率非常好
    // 改格式同时是列式存储的一种格式，所以对于SQL查询分析比较友好的
    val orcStuDF: DataFrame = spark
      .read
      .format("orc")
      .load("spark/data/stu/orc/")
    //    orcStuDF.show(1)

    // 4、parquet：同样在读取时不需要指定结构，同orc类似也会使用Snappy进行压缩
    // 相比orc，改格式更适合存储嵌套格式的数据
    val parquetStuDF: DataFrame = spark
      .read
      .format("parquet")
      .load("spark/data/stu/parquet/")
    parquetStuDF.show(2)

    // 5、如何加载MySQL中的数据
    val jdbcStuDF: DataFrame = spark
      .read
      .format("jdbc")
      .option("url", "jdbc:mysql://master:3306/student")
      .option("dbtable", "students")
      .option("user", "root")
      .option("password", "123456")
      .load()

    // 计算班级人数，写入MySQL


    // 第一种方式：使用RDD的方式
    val clazzCntRDD: RDD[(String, Int)] = jdbcStuDF
      .rdd
      .map(row => (row.getAs[String]("clazz"), 1))
      .reduceByKey(_ + _)

    val clazzCntDF01: DataFrame = clazzCntRDD
      .toDF("clazz", "cnt")

    /**
     * 修改MySQL默认的编码：
     * vim /etc/my.cnf
     *
     * [client]
     * default-character-set = utf8mb4
     * [mysqld]
     * character-set-server = utf8mb4
     * collation-server = utf8mb4_general_ci
     *
     * 重启MySQL服务：systemctl restart mysqld
     */
    clazzCntDF01
      .write
      .format("jdbc")
      //      .option("url", "jdbc:mysql://master:3306/student?useUnicode=true&characterEncoding=utf8")
      .option("url", "jdbc:mysql://master:3306/student")
      .option("dbtable", "clazz_cnt_spark")
      .option("user", "root")
      .option("password", "123456")
      .option("truncate", "true") // 以truncate的方式实现Overwrite，避免表结构发生变化
      // 指定写入模式
      .mode(SaveMode.Overwrite)
      // SparkSQL会默认将表删除，并按照Spark的类型映射关系建表，所以有可能会导致表结构发生变化
      .save()

    // 第二种方式：以Spark的SQL的方式完成
    // 需要先将DF注册成"表"
    jdbcStuDF.createOrReplaceTempView("jdbcStu")

    spark.sql(
        """
          |select clazz
          |       ,count(*) as cnt
          |from jdbcStu
          |group by clazz
          |""".stripMargin)
      // sql执行完之后也会返回一个DF，后续可以接show或者是write保存操作
      .show()

  }

}
